package com.shujia.flink.sink

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo3MySInk {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.readTextFile("data/students.txt")

    //使用自定义sink
    studentsDS.addSink(new MySink)

    env.execute()
  }
}

/**
 * 自定义斯诺克: 实现SInkSource接口
 */
class MySink extends SinkFunction[String] {
  /**
   * invoke： 每一条数据执行一次
   *
   * @param value   ： 一行数据
   * @param context ：上下文对象，可以获取flink的环境信息
   */
  override def invoke(value: String, context: SinkFunction.Context): Unit = {

    println(s"自定义sink:$value")
  }
}